package themis.flink.riskengine;

import com.alibaba.fastjson.JSONObject;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class RiskEngineMain {
    public static void main(String[] args) throws Exception {
        IRuleParseFactory strFactory = RuleManager.getParserFactory("str");
        IRuleParser ruleParser = strFactory.createRuleParser();
        List<Rule> rules = ruleParser.parseDSL("sum(amount#rcv_account.history,100000)");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        SourceFunction<String> source = null;
        DataStreamSource<String> streamSource = env.addSource(source);
        SingleOutputStreamOperator<JSONObject> singleOutputStreamOperator = streamSource.map(getMapper())
                .flatMap(new EventSplitFunction(rules))
                .keyBy(getKey_value())
                .map(new KeyEnrichFunction())
                .map(new FeatureEnrichFunction(rules))
                .keyBy(getEvent_id())
                .flatMap(new FeatureReduceFunction(rules))
                .map(new RuleBasedModeling());
        singleOutputStreamOperator.addSink(new PrintSinkFunction()).setParallelism(1);
        env.execute(RiskEngineMain.class.getSimpleName());
    }

    private static KeySelector<JSONObject, String> getEvent_id() {
        return new KeySelector<JSONObject, String>() {
            public String getKey(JSONObject value) throws Exception {
                return value.getString("EVENT_ID");
            }
        };
    }

    private static KeySelector<JSONObject, String> getKey_value() {
        return new KeySelector<JSONObject, String>() {
            public String getKey(JSONObject value) throws Exception {
                return value.getString("KEY_VALUE");
            }
        };
    }

    private static MapFunction<String, JSONObject> getMapper() {
        return new MapFunction<String, JSONObject>() {
            public JSONObject map(String s) throws Exception {
                if (StringUtils.isEmpty(s))
                    return new JSONObject();
                return JSONObject.parseObject(s);
            }
        };
    }
}
